package yunjiao.javatutorials.guava.concurrent.comprehensive;

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 *  线程池监控系统
 *
 * @author yangyunjiao
 */
public class MonitoredThreadPool {
    private final Lock lock = new ReentrantLock();
    private final ComprehensiveCondition taskAvailable;
    private final ComprehensiveCondition poolNotFull;
    private final Queue<Runnable> taskQueue;
    private final List<WorkerThread> workers;
    private final int maxPoolSize;
    private final int queueCapacity;
    private int activeWorkers = 0;
    private volatile boolean shutdown = false;

    public MonitoredThreadPool(int maxPoolSize, int queueCapacity) {
        this.maxPoolSize = maxPoolSize;
        this.queueCapacity = queueCapacity;
        this.taskQueue = new LinkedList<>();
        this.workers = new ArrayList<>();
        this.taskAvailable = new ComprehensiveCondition(lock.newCondition());
        this.poolNotFull = new ComprehensiveCondition(lock.newCondition());
    }

    public boolean execute(Runnable task, long timeout, TimeUnit unit) throws InterruptedException {
        if (shutdown) {
            throw new IllegalStateException("线程池已关闭");
        }

        lock.lock();
        try {
            // 如果队列已满，等待空间
            long nanosTimeout = unit.toNanos(timeout);
            while (taskQueue.size() >= queueCapacity && !shutdown) {
                if (nanosTimeout <= 0) {
                    return false;
                }
                nanosTimeout = poolNotFull.awaitNanos(nanosTimeout);
            }

            if (shutdown) {
                return false;
            }

            // 添加任务到队列
            taskQueue.offer(task);

            // 如果有空闲工作线程，唤醒一个；否则创建新线程（如果未达到最大限制）
            if (activeWorkers < maxPoolSize && getWaitingWorkerCount() == 0) {
                createWorker();
            } else {
                taskAvailable.signal();
            }

            return true;

        } finally {
            lock.unlock();
        }
    }

    public void shutdown() {
        lock.lock();
        try {
            shutdown = true;
            // 唤醒所有工作线程
            taskAvailable.signalAll();
            // 等待所有工作线程完成
            for (WorkerThread worker : workers) {
                try {
                    worker.join(1000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        } finally {
            lock.unlock();
        }

        printThreadPoolStatistics();
    }

    public void shutdownNow() {
        lock.lock();
        try {
            shutdown = true;
            taskQueue.clear();
            // 中断所有工作线程
            for (WorkerThread worker : workers) {
                worker.interrupt();
            }
            taskAvailable.signalAll();
        } finally {
            lock.unlock();
        }

        printThreadPoolStatistics();
    }

    private void createWorker() {
        WorkerThread worker = new WorkerThread();
        workers.add(worker);
        activeWorkers++;
        worker.start();
        System.out.printf("创建新工作线程，活跃线程数: %d/%d%n", activeWorkers, maxPoolSize);
    }

    private int getWaitingWorkerCount() {
        int count = 0;
        for (WorkerThread worker : workers) {
            if (worker.isWaiting()) {
                count++;
            }
        }
        return count;
    }

    public void printThreadPoolStatistics() {
        System.out.println("\n" + "=".repeat(70));
        System.out.println("                   线程池详细统计报告");
        System.out.println("=".repeat(70));

        System.out.printf("线程池状态: %s%n", shutdown ? "已关闭" : "运行中");
        System.out.printf("线程统计: %d/%d (活跃/最大)%n", activeWorkers, maxPoolSize);
        System.out.printf("队列统计: %d/%d (当前/容量)%n", taskQueue.size(), queueCapacity);
        System.out.printf("等待中的工作线程: %d%n", getWaitingWorkerCount());

        System.out.println("\n任务可用条件统计:");
        taskAvailable.printDetailedStatistics();

        System.out.println("\n池未满条件统计:");
        poolNotFull.printDetailedStatistics();

        // 计算线程池效率指标
        double queueUtilization = (double) taskQueue.size() / queueCapacity * 100;
        double workerUtilization = (double) activeWorkers / maxPoolSize * 100;
        System.out.printf("队列利用率: %.1f%%%n", queueUtilization);
        System.out.printf("线程利用率: %.1f%%%n", workerUtilization);

        System.out.println("=".repeat(70));
    }

    private class WorkerThread extends Thread {
        private volatile boolean waiting = false;

        public WorkerThread() {
            super("Worker-" + (workers.size() + 1));
        }

        public boolean isWaiting() {
            return waiting;
        }

        @Override
        public void run() {
            System.out.println(getName() + ": 开始运行");

            while (!shutdown && !isInterrupted()) {
                Runnable task = null;

                lock.lock();
                try {
                    while (taskQueue.isEmpty() && !shutdown && !isInterrupted()) {
                        waiting = true;
                        taskAvailable.await();
                        waiting = false;
                    }

                    if (shutdown && taskQueue.isEmpty()) {
                        break;
                    }

                    task = taskQueue.poll();

                    // 通知可能有空间添加新任务
                    if (taskQueue.size() == queueCapacity - 1) {
                        poolNotFull.signal();
                    }

                } catch (InterruptedException e) {
                    System.out.println(getName() + ": 被中断");
                    break;
                } finally {
                    lock.unlock();
                }

                if (task != null) {
                    try {
                        System.out.println(getName() + ": 执行任务");
                        task.run();
                        System.out.println(getName() + ": 任务完成");
                    } catch (Exception e) {
                        System.err.println(getName() + ": 任务执行失败 - " + e.getMessage());
                    }
                }
            }

            activeWorkers--;
            System.out.println(getName() + ": 结束运行，剩余活跃线程: " + activeWorkers);
        }
    }

    public static void main(String[] args) throws InterruptedException {
        MonitoredThreadPool threadPool = new MonitoredThreadPool(3, 10);

        // 提交任务
        for (int i = 1; i <= 20; i++) {
            final int taskId = i;
            boolean submitted = threadPool.execute(() -> {
                System.out.println("执行任务 " + taskId);
                try {
                    Thread.sleep(200); // 模拟任务执行时间
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }, 1, TimeUnit.SECONDS);

            if (!submitted) {
                System.out.println("任务 " + taskId + " 提交失败");
            }

            Thread.sleep(50); // 控制任务提交速率
        }

        // 让线程池运行一会儿
        Thread.sleep(2000);

        // 优雅关闭
        threadPool.shutdown();
    }
}
